package com.polaris.live.im.sfs.core

import com.polaris.live.common.cacha.CacheUtil
import com.polaris.live.common.network.ApiResponse
import com.polaris.live.common.network.AppException
import com.polaris.live.common.network.HTTP_ERROR_CODE
import com.polaris.live.common.network.HTTP_SUCCESS_CODE
import com.polaris.live.common.util.JsonUtils
import com.polaris.live.common.util.computeIfAbsentBy21
import com.polaris.live.im.sfs.core.bean.AbstractMessage
import com.polaris.live.im.sfs.core.bean.DataArgument
import com.polaris.live.im.sfs.core.bean.EventWrapper
import com.polaris.live.im.sfs.core.bean.ExtractMessageResponse
import com.polaris.live.im.sfs.core.bean.MessageRequest
import com.polaris.live.im.sfs.core.bean.MessageResponse
import com.polaris.live.im.sfs.core.bean.ReceiptMessage
import com.polaris.live.im.sfs.core.bean.WorkDeferred
import com.polaris.live.im.sfs.core.bean.impl.AbstractChatMessageRequest
import com.polaris.live.im.sfs.core.bean.impl.AbstractLiveMessageRequest
import com.polaris.live.im.sfs.core.bean.impl.AbstractPusherMessageRequest
import com.polaris.live.im.sfs.core.bean.impl.DefaultChatMessageRequest
import com.polaris.live.im.sfs.core.bean.impl.DefaultLiveMessageRequest
import com.polaris.live.im.sfs.core.bean.impl.DefaultPusherMessageRequest
import com.polaris.live.im.sfs.core.bean.resolveResponse
import com.polaris.live.im.sfs.interfaces.OnImMessageListener
import com.polaris.live.im.sfs.processor.impl.SingleMessageMessageProcessor
import com.polaris.live.im.sfs.processor.impl.system.EventDeferredMessageProcessor
import com.polaris.live.im.sfs.processor.impl.system.EventListen
import com.polaris.live.im.sfs.processor.impl.system.ReceiptMessageProcessor
import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.Deferred
import kotlinx.coroutines.withTimeout
import sfs2x.client.entities.Room
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.CopyOnWriteArrayList
import kotlin.reflect.typeOf

/**
 * 消息队列管理器
 *
 * 负责发送和接收消息
 *
 * @author Created by lucas on 2023/10/18 19:50
 */
object SmartFoxQueueManager : OnImMessageListener {

    private const val TIMEOUT_SEND = 15 * 1000L

    private val workQueue = ConcurrentHashMap<String, WorkDeferred>()
    private val chatWaitQueue = ConcurrentHashMap<MessageRequest, CompletableDeferred<ReceiptMessage>>()
    private val liveWaitQueue = ConcurrentHashMap<MessageRequest, CompletableDeferred<ReceiptMessage>>()
    private val pusherWaitQueue = ConcurrentHashMap<MessageRequest, CompletableDeferred<ReceiptMessage>>()

    private val eventQueue = ConcurrentHashMap<String, MutableList<EventWrapper>>()

    private val messageMediator: MessageMediator

    @Volatile
    private var isChatConnected = false

    @Volatile
    private var isLiveConnected = false

    @Volatile
    private var isPusherConnected = false

    private val singleMessageMessageProcessor = SingleMessageMessageProcessor()

    init {
        messageMediator = MessageMediator(
            listOf(
                ReceiptMessageProcessor(workQueue),
                EventDeferredMessageProcessor(eventQueue),
                singleMessageMessageProcessor
            )
        )
    }

    fun onConnected() {
        //ignore
    }

    fun onDisconnected() {
        //ignore
    }

    fun onChatConnected() {
        isChatConnected = true
        handleWaitQueue(SmartFoxConnector.CLIENT_CHAT, chatWaitQueue)

        //连接成功后
        singleMessageMessageProcessor.fetchUnreadSingleMessage()
    }

    fun onChatDisconnected() {
        isChatConnected = false
    }

    fun onLiveConnected() {
        isLiveConnected = true
        handleWaitQueue(SmartFoxConnector.CLIENT_LIVE, liveWaitQueue)
    }

    fun onLiveDisconnected() {
        isLiveConnected = false
    }

    fun onPusherConnected() {
        isPusherConnected = true
        handleWaitQueue(SmartFoxConnector.CLIENT_PUSHER, pusherWaitQueue)
    }

    fun onPusherDisconnected() {
        isPusherConnected = false
    }

    fun onChatEnterRoom(roomId: String) {
        handleWaitQueue(SmartFoxConnector.CLIENT_CHAT, roomId, chatWaitQueue)
    }

    fun onLiveEnterRoom(roomId: String) {
        handleWaitQueue(SmartFoxConnector.CLIENT_LIVE, roomId, liveWaitQueue)
    }

    fun onPusherEnterRoom(roomId: String) {
        handleWaitQueue(SmartFoxConnector.CLIENT_PUSHER, roomId, pusherWaitQueue)
    }

    fun isWorking(): Boolean {
        return workQueue.isNotEmpty()
    }

    private fun handleWaitQueue(type: Int, waitQueue: MutableMap<MessageRequest, CompletableDeferred<ReceiptMessage>>) {
        handleWaitQueue(type, null, waitQueue)
    }

    private fun handleWaitQueue(
        type: Int,
        roomId: String?,
        waitQueue: MutableMap<MessageRequest, CompletableDeferred<ReceiptMessage>>,
    ) {
        val iterator = waitQueue.entries.iterator()
        while (iterator.hasNext()) {
            val (message, deferred) = iterator.next()
            //作用域为room的消息在未加入房间之前不能发送
            if (message.reqRoom != null) {
                if (roomId == null || roomId != message.reqRoom.name) {
                    continue
                }
            }

            workQueue[message.msgId] = WorkDeferred(message, deferred)
            SmartFoxManager.sendMessage(type, message)

            iterator.remove()
        }
    }

    override suspend fun onMessage(command: String, arguments: Map<String, Any>) {
        messageMediator.onMessage(command, arguments)
    }

    private fun addQueue(type: Int, message: MessageRequest): Deferred<ReceiptMessage> {
        return CompletableDeferred<ReceiptMessage>().also {
            if (type == SmartFoxConnector.CLIENT_CHAT && !isChatConnected) {
                chatWaitQueue[message] = it
            } else if (type == SmartFoxConnector.CLIENT_LIVE && !isLiveConnected) {
                liveWaitQueue[message] = it
            } else if (type == SmartFoxConnector.CLIENT_PUSHER && !isPusherConnected) {
                pusherWaitQueue[message] = it
            } else {
                workQueue[message.msgId] = WorkDeferred(message, it)
                SmartFoxManager.sendMessage(type, message)
            }
            it.invokeOnCompletion {
                when (type) {
                    SmartFoxConnector.CLIENT_CHAT -> chatWaitQueue.remove(message)
                    SmartFoxConnector.CLIENT_LIVE -> liveWaitQueue.remove(message)
                    SmartFoxConnector.CLIENT_PUSHER -> pusherWaitQueue.remove(message)
                }
            }
        }
    }

    /**
     * 发送消息
     */
    suspend fun <R : MessageRequest> sendMessage(message: R, timeout: Long = TIMEOUT_SEND): MessageResponse<R> {
        val type = when (message) {
            is AbstractChatMessageRequest -> SmartFoxConnector.CLIENT_CHAT
            is AbstractLiveMessageRequest -> SmartFoxConnector.CLIENT_LIVE
            is AbstractPusherMessageRequest -> SmartFoxConnector.CLIENT_PUSHER
            else -> throw IllegalArgumentException("unsupported message type")
        }
        return withTimeout(timeout) {
            val receipt = addQueue(type, message).await()

            MessageResponse(message, receipt)
        }
    }

    /**
     * 发送私聊消息
     */
    suspend fun <M : AbstractMessage> sendChatMessage(
        command: String,
        message: M,
        room: Room? = null,
        timeout: Long = TIMEOUT_SEND,
    ): ExtractMessageResponse<M> {
        val request = DefaultChatMessageRequest(message.msgId, command, message, room)
        val response = sendMessage(request, timeout)
        return ExtractMessageResponse(message, response.receipt, response.receipt.ext)
    }

    /**
     * 发送直播消息
     */
    suspend fun <M : AbstractMessage> sendLiveMessage(
        command: String,
        message: M,
        room: Room? = null,
        timeout: Long = TIMEOUT_SEND,
    ): ExtractMessageResponse<M> {
        val request = DefaultLiveMessageRequest(message.msgId, command, message, room)
        val response = sendMessage(request, timeout)
        return ExtractMessageResponse(message, response.receipt, response.receipt.ext)
    }

    /**
     * 发送推送消息
     */
    suspend fun <M : AbstractMessage> sendPusherMessage(
        command: String,
        message: M,
        room: Room? = null,
        timeout: Long = TIMEOUT_SEND,
    ): ExtractMessageResponse<M> {
        val request = DefaultPusherMessageRequest(message.msgId, command, message, room)
        val response = sendMessage(request, timeout)
        return ExtractMessageResponse(message, response.receipt, response.receipt.ext)
    }

    /**
     * 监听事件
     *
     * @return data尽量快速处理，外部不允许存储
     */
    suspend fun eventListen(
        @ClientType type: Int,
        @EventListen eventType: String,
        timeout: Long = TIMEOUT_SEND,
        condition: OnEventCondition? = null,
    ): DataArgument {
        return withTimeout(timeout) {
            val deferred = CompletableDeferred<DataArgument>()
            val wrapper = EventWrapper(type, eventType, deferred, condition)

            val list = eventQueue.computeIfAbsentBy21(eventType) { CopyOnWriteArrayList() }

            list.add(wrapper)

            deferred.invokeOnCompletion {
                list.remove(wrapper)
            }
            deferred.await()
        }
    }

    /**
     * 监听直播事件
     *
     * @return data尽量快速处理，外部不允许存储
     */
    suspend fun eventLiveListen(
        @EventListen eventType: String,
        timeout: Long = TIMEOUT_SEND,
        condition: OnEventCondition? = null,
    ): DataArgument {
        return eventListen(SmartFoxConnector.CLIENT_LIVE, eventType, timeout, condition)
    }
}

fun interface OnEventCondition {

    fun test(data: DataArgument): Boolean

}

/**
 * 转换为ApiResponse
 */
inline fun <reified T : Any> withMessage(
    block: () -> ExtractMessageResponse<AbstractMessage>,
): ApiResponse<T> {
    val timestamp = CacheUtil.getServerTime()

    return try {
        val message = block()

        val extMap = message.ext?.let {
            JsonUtils.fromJson<Map<String, Any>>(it)
        }

        val response = if (typeOf<T>() == Unit::class) {
            Unit as T
        } else {
            message.receipt.resolveResponse()
        }?.let {
            ApiResponse(HTTP_SUCCESS_CODE, timestamp, null, extMap, it)
        }

        response ?: throw NullPointerException("Unable to parse response")
    } catch (t: Throwable) {
        if (t is AppException) {
            ApiResponse(t.errCode, timestamp, t.message, t.ext, null)
        } else {
            ApiResponse(HTTP_ERROR_CODE, timestamp, t.message, null, null)
        }
    }
}